package com.zoe.onelink.logging.trace.zipkin.interceptor;

import cn.hutool.core.util.ZipUtil;
import com.zoe.onelink.logging.trace.property.OnelinkZipkinProperties;
import com.zoe.onelink.logging.trace.zipkin.util.ZipkinSpanUtil;
import com.zoe.onelink.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.sleuth.zipkin2.ZipkinProperties;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.ClientHttpRequestExecution;
import org.springframework.http.client.ClientHttpRequestInterceptor;
import org.springframework.http.client.ClientHttpResponse;
import zipkin2.codec.SpanBytesDecoder;
import zipkin2.codec.SpanBytesEncoder;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Date;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * <p>标题: </p>
 * <p>描述: </p>
 * <p>版权: Copyright (c) 2022</p>
 * <p>公司: 智业软件股份有限公司</p>
 *
 * @version: 1.0
 * @author: ruansheng
 * @date: 2022-10-10
 */
@Slf4j
public class ZipkinRestTemplateLoggingInterceptor implements ClientHttpRequestInterceptor {

    private final ZipkinProperties zipkinProperties;
    private final OnelinkZipkinProperties onelinkZipkinProperties;
    private final SpanBytesDecoder spanBytesDecoder;

    /**
     * 熔断标记
     */
    private final AtomicBoolean isFused = new AtomicBoolean(false);

    /**
     * 熔断阈值
     */
    private final AtomicInteger httpErrorCount = new AtomicInteger(0);

    /**
     * 最后熔断时间
     */
    private final AtomicLong lastFusingTime = new AtomicLong();

    private final AtomicInteger errorCount = new AtomicInteger();

    public ZipkinRestTemplateLoggingInterceptor(ZipkinProperties zipkinProperties, OnelinkZipkinProperties onelinkZipkinProperties) {
        this.zipkinProperties = zipkinProperties;
        this.onelinkZipkinProperties = onelinkZipkinProperties;
        this.spanBytesDecoder = this.getSpanBytesDecoder(zipkinProperties);
    }

    private SpanBytesDecoder getSpanBytesDecoder(ZipkinProperties zipkinProperties) {
        SpanBytesEncoder encoder = zipkinProperties.getEncoder();
        if (encoder == null) {
            return SpanBytesDecoder.JSON_V2;
        }
        try {
            return SpanBytesDecoder.valueOf(encoder.name());
        } catch (Exception e) {
            return SpanBytesDecoder.JSON_V2;
        }
    }

    @Override
    public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException {
        ClientHttpResponse response;
        // 检查是否进行熔断
        this.checkFusing(request, body);
        try {
            response = execution.execute(request, body);
            // 重置熔断状态
            this.resetFusingStatus();
            // todo 如果距离上次熔断时间超过一定时间,则清空错误次数
        } catch (Exception e) {
            // 更新熔断状态
            this.updateFusingStatusOnError();
            // 记录发送失败的链路日志
            this.logSpan(request, body);
            throw e;
        }
        return response;
    }

    /**
     * 检查是否进行熔断
     */
    private void checkFusing(HttpRequest request, byte[] body) throws InterruptedIOException {
        // 已熔断
        if (this.onelinkZipkinProperties.isFusingEnabled() && this.isFused.get()) {
            // 熔断时间
            long currentFusingTime = System.currentTimeMillis() - this.lastFusingTime.get();
            // 熔断时间超出设定的范围时,重置熔断状态
            if (currentFusingTime > this.onelinkZipkinProperties.getFusingTime().toMillis()) {
                this.resetFusingStatus();
            } else {
                this.logSpan(request, body);
                throw new InterruptedIOException("Zipkin restTemplate is fused,last fuse time is " + DateUtil.formatDateTime(new Date(this.lastFusingTime.get())));
            }
        }
    }

    /**
     * 更新熔断状态
     */
    private synchronized void updateFusingStatusOnError() {
        // 未开启熔断或已是熔断状态直接返回
        if (!this.onelinkZipkinProperties.isFusingEnabled() || this.isFused.get()) {
            return;
        }
        // 非熔断状态且错误数未达到阈值
        if (this.httpErrorCount.incrementAndGet() < this.onelinkZipkinProperties.getFusingThreshold()) {
            return;
        }
        log.warn("[ZipkinTrace] 链路发送错误次数达到阈值:{},熔断机制已启动,熔断时间:{}",
                this.onelinkZipkinProperties.getFusingThreshold(), this.onelinkZipkinProperties.getFusingTime());
        this.lastFusingTime.set(System.currentTimeMillis());
        this.isFused.set(true);
    }

    /**
     * 重置熔断状态
     */
    private void resetFusingStatus() {
        if (this.isFused.compareAndSet(true, false)) {
            log.info("[ZipkinTrace] 重置熔断状态");
            this.httpErrorCount.set(0);
        }
    }

    private void logSpan(HttpRequest request, byte[] body) {
        try {
            byte[] rawBody;
            if (zipkinProperties.getCompression().isEnabled() && ZipkinRestTemplateGZipInterceptor.CONTENT_ENCODING_TYPE.equals(request.getHeaders().getFirst(ZipkinRestTemplateGZipInterceptor.CONTENT_ENCODING_HEADER_NAME))) {
                rawBody = ZipUtil.unGzip(body);
            } else {
                rawBody = body;
            }
            ZipkinSpanUtil.logSpan(rawBody, this.spanBytesDecoder);
        } catch (Exception e) {
            if (this.errorCount.incrementAndGet() == 1000) {
                this.errorCount.set(0);
                log.error("[ZipkinTrace] 链路日志发送失败", e);
            }
        }
    }

}
